package com.alibaba.datax.plugin.reader.aliyunslsreader.util;

import com.alibaba.datax.common.element.*;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.reader.aliyunslsreader.AliyunSlsReaderErrorCode;
import com.alibaba.datax.plugin.unstructuredstorage.reader.ColumnEntry;
import com.alibaba.datax.plugin.unstructuredstorage.reader.Key;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderErrorCode;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.text.DateFormat;
import java.util.Date;
import java.util.List;

/**
 * Created by dancoder on 2022.01.08
 */
public class LogHubReaderUtil {

    private static final Logger LOG = LoggerFactory.getLogger(Reader.Task.class);

    public static Record getRecord(RecordSender recordSender, Configuration configuration, JSONObject oneRecordJson, TaskPluginCollector taskPluginCollector) {
        Record record = recordSender.createRecord();

        List<ColumnEntry> columns = UnstructuredStorageReaderUtil.getListColumnEntry(configuration, Key.COLUMN);
        for (ColumnEntry columnEntry : columns) {
            String columnName = columnEntry.getValue();
            String columnValue = oneRecordJson.getString(columnName);

            System.out.printf("type:%s,\t value:%s, \t format:%s%n", columnEntry.getType(), columnName, columnEntry.getFormat());

            if (! oneRecordJson.containsKey(columnName)) {
                throw DataXException.asDataXException(AliyunSlsReaderErrorCode.SLS_READER_CONFIG_ERROR,
                        String.format("您配置的column里不存在实际消费时获取到的column，请检查project、logstore 或者 column是否配置正确。不存在的column：[%s]", columnEntry.getValue()));
            }

            Column column = null;
            Column.Type type = null;

            String columnType = columnEntry.getType();
            try {
                type = Column.Type.valueOf(columnType.toUpperCase());
            } catch (IllegalArgumentException e) {
                throw DataXException.asDataXException(AliyunSlsReaderErrorCode.SLS_READER_CONFIG_ERROR, String.format("您配置的列类型暂不支持 : [%s]", columnType));
            }

            try {
                switch (type) {
                    case STRING:
                        column = new StringColumn(columnValue);
                        break;
                    case BYTES:
                        try {
                            column = new BytesColumn(columnValue.getBytes(StandardCharsets.UTF_8));
                        } catch (Exception e) {
                            throw DataXException.asDataXException(AliyunSlsReaderErrorCode.SLS_READER_CONFIG_ERROR, String.format("类型转换错误, 无法将[%s] 转换为[%s]", columnValue, "BYTES"));
                        }
                        break;
                    case INT:
                    case LONG:
                        try {
                            column = new LongColumn(columnValue);
                        } catch (Exception e) {
                            throw DataXException.asDataXException(AliyunSlsReaderErrorCode.SLS_READER_CONFIG_ERROR, String.format("类型转换错误, 无法将[%s] 转换为[%s]", columnValue, "LONG"));
                        }
                        break;
                    case DOUBLE:
                        try {
                            column = new DoubleColumn(columnValue);
                        } catch (Exception e) {
                            throw DataXException.asDataXException(AliyunSlsReaderErrorCode.SLS_READER_CONFIG_ERROR, String.format("类型转换错误, 无法将[%s] 转换为[%s]", columnValue, "DOUBLE"));
                        }
                        break;
                    case BOOL:
                        try {
                            column = new BoolColumn(columnValue);
                        } catch (Exception e) {
                            throw DataXException.asDataXException(AliyunSlsReaderErrorCode.SLS_READER_CONFIG_ERROR, String.format("类型转换错误, 无法将[%s] 转换为[%s]", columnValue, "BOOLEAN"));
                        }
                        break;
                    case DATE:
                        try {
                            if (columnValue == null) {
                                Date date = null;
                                column = new DateColumn(date);
                            } else {
                                String formatString = columnEntry.getFormat();
                                if (StringUtils.isNotBlank(formatString)) {
                                    // 用户自己配置的格式转换, 脏数据行为出现变化
                                    DateFormat format = columnEntry.getDateFormat();
                                    column = new DateColumn(format.parse(columnValue));
                                } else {
                                    // 框架尝试转换
                                    column = new DateColumn(new StringColumn(columnValue).asDate());
                                }
                            }
                        } catch (Exception e) {
                            throw DataXException.asDataXException(AliyunSlsReaderErrorCode.SLS_READER_CONFIG_ERROR, String.format("类型转换错误, 无法将[%s] 转换为[%s]", columnValue, "DATE"));
                        }
                        break;
                    default:
                        throw DataXException.asDataXException(UnstructuredStorageReaderErrorCode.NOT_SUPPORT_TYPE, String.format("您配置的列类型暂不支持 : [%s]", columnType));
                }
            } catch (IllegalArgumentException | IndexOutOfBoundsException iae) {
                taskPluginCollector.collectDirtyRecord(record, iae.getMessage());
            } catch (Exception e) {
                if (e instanceof DataXException) {
                    throw (DataXException) e;
                }
                // 每一种转换失败都是脏数据处理,包括数字格式 & 日期格式
                taskPluginCollector.collectDirtyRecord(record, e.getMessage());
            }


            System.out.printf("content:%s, \t type:%s%n", column.getRawData(), column.getType());
            record.addColumn(column);
            System.out.println("=================================================转换完成=======================================");
        }
        return record;
    }
}